package com.by.order.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.by.common.constant.OrderConstant;
import com.by.common.exception.ErrorCode;
import com.by.order.api.service.IReliableMessageService;
import com.by.order.api.vo.ReliableMessageVo;
import com.by.order.entity.ReliableMessageEntity;
import com.by.order.mapper.ReliableMessageMapper;
import com.by.rpc.exception.BusinessException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;

/**
 * 可靠消息服务实现类
 */
@Slf4j
@Service
public class ReliableMessageServiceImpl implements IReliableMessageService {

    @Autowired
    private ReliableMessageMapper reliableMessageMapper;

    /**
     * 在本地事务中记录一条待发送的消息，并返回消息ID。
     *
     * @param messageContent 消息内容
     * @param businessId     业务标识，例如订单ID
     * @param messageType    消息类型，用于消费者识别处理
     * @return 记录成功的消息ID
     */
    @Override
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public String recordMessageInLocalTx(String messageContent, String businessId, String messageType) {
        ReliableMessageEntity message = new ReliableMessageEntity();
        String messageId = UUID.randomUUID().toString();
        message.setMessageId(messageId);
        message.setMessageBody(messageContent);
        message.setStatus(OrderConstant.MESSAGE_STATUS_TO_BE_SENT); //1 :待发送
        message.setMessageType(messageType);
        message.setBusinessId(businessId);
        message.setTryCount(0);
        message.setCreateTime(LocalDateTime.now());
        message.setUpdateTime(LocalDateTime.now());

        int result = reliableMessageMapper.insert(message);
        if (result != 1) {
            log.error("Failed to insert reliable message. BusinessId: {}", businessId);
            throw new BusinessException(ErrorCode.SYSTEM_BUSY.getCode(), "Failed to record reliable message");
        }
        log.debug("Successfully recorded reliable message with ID: {}", messageId);
        return messageId;
    }

    /**
     * 根据业务ID查找对应的可靠消息记录。
     *
     * @param businessId 业务标识
     * @return 消息VO对象，如果不存在则返回null
     */
    @Override
    public ReliableMessageVo getMessageByBusinessId(String businessId) {
        ReliableMessageEntity messageEntity = reliableMessageMapper.selectOne(new LambdaQueryWrapper<ReliableMessageEntity>()
                .eq(ReliableMessageEntity::getBusinessId, businessId)
        );
        if (messageEntity == null) {
            return null;
        }
        ReliableMessageVo messageVo = new ReliableMessageVo();
        BeanUtils.copyProperties(messageEntity, messageVo);
        return messageVo;
    }

    /**
     * 根据消息ID查找对应的可靠消息记录。
     *
     * @param messageId 消息ID
     * @return 消息VO对象，如果不存在则返回null
     */
    @Override
    public ReliableMessageVo getMessageById(String messageId) {
        ReliableMessageEntity messageEntity = reliableMessageMapper.selectOne(new LambdaQueryWrapper<ReliableMessageEntity>()
                .eq(ReliableMessageEntity::getMessageId, messageId)
        );
        if (messageEntity == null) {
            return null;
        }
        ReliableMessageVo messageVo = new ReliableMessageVo();
        BeanUtils.copyProperties(messageEntity, messageVo);
        return messageVo;
    }

    /**
     * 获取指定数量和状态的待处理消息列表。
     *
     * @param status 消息状态
     * @param limit  查询数量
     * @return 待处理的消息VO列表
     */
    @Override
    public List<ReliableMessageVo> getMessagesByStatus(int status, int limit) {
        List<ReliableMessageEntity> messages = reliableMessageMapper.selectList(new LambdaQueryWrapper<ReliableMessageEntity>()
                .eq(ReliableMessageEntity::getStatus, status)
                .last("LIMIT " + limit)
        );

        return messages.stream()
                .map(entity -> {
                    ReliableMessageVo vo = new ReliableMessageVo();
                    BeanUtils.copyProperties(entity, vo);
                    return vo;
                })
                .collect(Collectors.toList());
    }

    /**
     * 更新消息状态。
     *
     * @param messageId 消息ID
     * @param newStatus 新状态
     * @param tryCount  尝试次数
     * @return 是否更新成功
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean updateMessageStatus(String messageId, int newStatus, int tryCount) {
        LambdaUpdateWrapper<ReliableMessageEntity> updateWrapper = new LambdaUpdateWrapper<>();
        updateWrapper.eq(ReliableMessageEntity::getMessageId, messageId)
                .set(ReliableMessageEntity::getStatus, newStatus)
                .set(ReliableMessageEntity::getTryCount, tryCount)
                .set(ReliableMessageEntity::getUpdateTime, LocalDateTime.now());

        int result = reliableMessageMapper.update(null, updateWrapper);
        return result == 1;
    }

    /**
     * 增加消息尝试次数。
     *
     * @param messageId 消息ID
     * @return 是否更新成功
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean incrementTryCount(String messageId) {
        LambdaUpdateWrapper<ReliableMessageEntity> updateWrapper = new LambdaUpdateWrapper<>();
        updateWrapper.eq(ReliableMessageEntity::getMessageId, messageId)
                .setSql("try_count = try_count + 1")
                .set(ReliableMessageEntity::getUpdateTime, LocalDateTime.now());

        int result = reliableMessageMapper.update(null, updateWrapper);
        return result == 1;
    }
}